package endpoint

import (
	"context"
	"github.com/darwinOrg/go-monitor"
	"github.com/go-mysql-org/go-mysql/canal"
	"github.com/go-mysql-org/go-mysql/mysql"
	"github.com/juju/errors"
	client "github.com/zinclabs/sdk-go-zincsearch"
	"log"
	"myc/global"
	"myc/metrics"
	"myc/model"
	"myc/service/luaengine"
	"myc/util/alarm"
	"myc/util/logs"
	"myc/util/stringutil"
)

const (
	zincsearchPingErrorCounter = "zincsearch_ping_error"
)

var (
	zincsearchPingErrorLabelMap = map[string]string{"error": zincsearchPingErrorCounter}
)

type ZincSearchEndpoint struct {
	host   string
	client *client.APIClient
}

func newZincSearchEndpoint() *ZincSearchEndpoint {
	host := global.Cfg().ZincAddr
	r := &ZincSearchEndpoint{}
	r.host = host
	return r
}

func (s *ZincSearchEndpoint) Name() string {
	return "zincsearch"
}

func (s *ZincSearchEndpoint) Connect() error {
	configuration := client.NewConfiguration()
	configuration.Servers = client.ServerConfigurations{
		client.ServerConfiguration{
			URL: s.host,
		},
	}

	s.client = client.NewAPIClient(configuration)
	return nil
}

func (s *ZincSearchEndpoint) Ping() error {
	if _, _, err := s.client.Default.Healthz(context.Background()).Execute(); err == nil {
		return nil
	}

	err := monitor.IncCounter(zincsearchPingErrorCounter, zincsearchPingErrorLabelMap)
	if err != nil {
		logs.Warnf("monitor.IncCounter(%s) error: %v", zincsearchPingErrorCounter, err)
	}

	return errors.New("ssx")
}

func (s *ZincSearchEndpoint) Consume(from mysql.Position, rows []*model.RowRequest) error {
	bulk := s.client.Document.Bulkv2(context.Background())
	docsMap := map[string][]map[string]interface{}{}

	for _, row := range rows {
		rule, _ := global.RuleIns(row.RuleKey)
		if rule.TableColumnSize != len(row.Row) {
			logs.Warnf("%s schema mismatching", row.RuleKey)
			continue
		}

		metrics.UpdateActionNum(row.Action, row.RuleKey)

		if rule.LuaEnable() {
			kvm := rowMap(row, rule, true)
			ls, err := luaengine.DoESOps(kvm, row.Action, rule)
			if err != nil {
				log.Println("Lua 脚本执行失败!!! ,详情请参见日志")
				alarm.EmergencyAlarm(err)
				return errors.Errorf("lua 脚本执行失败 : %s ", errors.ErrorStack(err))
			}
			for _, resp := range ls {
				logs.Infof("action: %s, Index: %s , Id:%s, value: %v", resp.Action, resp.Index, resp.Id, resp.Date)
				var doc map[string]interface{}
				err = json.Unmarshal([]byte(resp.Date), &doc)
				if err != nil {
					alarm.EmergencyAlarm(err)
					return err
				}

				if resp.Action == canal.InsertAction || resp.Action == canal.UpdateAction {
					docsMap[rule.ZincIndex] = append(docsMap[rule.ZincIndex], doc)
				} else if resp.Action == canal.DeleteAction {
					_, _, err = s.client.Document.Delete(context.Background(), rule.ZincIndex, resp.Id).Execute()
					if err != nil {
						logs.Errorf("删除文档失败, index: %s, id: %s, err: %s", rule.ZincIndex, resp.Id, errors.ErrorStack(err))
						alarm.EmergencyAlarm(err)
						return err
					}
				}
			}
		} else {
			kvm := rowMap(row, rule, false)
			id := primaryKey(row, rule)
			strId := stringutil.ToString(id)
			kvm["_id"] = strId
			body := encodeValue(rule, kvm)
			logs.Infof("action: %s, Index: %s , Id:%s, value: %v", row.Action, rule.ZincIndex, strId, body)
			if row.Action == canal.InsertAction || row.Action == canal.UpdateAction {
				docsMap[rule.ZincIndex] = append(docsMap[rule.ZincIndex], kvm)
			} else if row.Action == canal.DeleteAction {
				_, _, err := s.client.Document.Delete(context.Background(), rule.ZincIndex, strId).Execute()
				if err != nil {
					logs.Errorf("删除文档失败, index: %s, id: %s, err: %s", rule.ZincIndex, strId, errors.ErrorStack(err))
					alarm.EmergencyAlarm(err)
					return err
				}
			}
		}
	}

	if len(docsMap) > 0 {
		for index, docs := range docsMap {
			resp, _, err := bulk.Query(client.MetaJSONIngest{
				Index:   &index,
				Records: docs,
			}).Execute()
			if err != nil {
				ifZincsearchOpenapiErrLogBody(err)
				alarm.EmergencyAlarm(err)
				return err
			}
			logs.Infof("index: %s, resp message: %s, recourdCount: %d", index, *resp.Message, *resp.RecordCount)
		}
	}

	logs.Infof("%s 处理完成 %d 条数据", s.Name(), len(rows))
	return nil
}

func (s *ZincSearchEndpoint) Stock(rows []*model.RowRequest) int64 {
	if len(rows) == 0 {
		return 0
	}

	bulk := s.client.Document.Bulkv2(context.Background())
	docsMap := map[string][]map[string]interface{}{}
	var count int64

	for _, row := range rows {
		rule, _ := global.RuleIns(row.RuleKey)
		if rule.TableColumnSize != len(row.Row) {
			logs.Warnf("%s schema mismatching", row.RuleKey)
			continue
		}

		if rule.LuaEnable() {
			kvm := rowMap(row, rule, true)
			ls, err := luaengine.DoESOps(kvm, row.Action, rule)
			if err != nil {
				logs.Errorf("lua 脚本执行失败 : %s ", errors.ErrorStack(err))
				alarm.EmergencyAlarm(err)
				break
			}
			for _, resp := range ls {
				var doc map[string]interface{}
				err = json.Unmarshal([]byte(resp.Date), &doc)
				if err != nil {
					alarm.EmergencyAlarm(err)
					return 0
				}
				if resp.Action == canal.InsertAction || resp.Action == canal.UpdateAction {
					docsMap[rule.ZincIndex] = append(docsMap[rule.ZincIndex], doc)
				} else if resp.Action == canal.DeleteAction {
					_, _, err = s.client.Document.Delete(context.Background(), rule.ZincIndex, resp.Id).Execute()
					if err != nil {
						logs.Errorf("删除文档失败, index: %s, id: %s, err: %s", rule.ZincIndex, resp.Id, errors.ErrorStack(err))
						alarm.EmergencyAlarm(err)
						return count
					}
					count++
				}
			}
		} else {
			kvm := rowMap(row, rule, false)
			id := primaryKey(row, rule)
			strId := stringutil.ToString(id)
			kvm["_id"] = strId
			//body := encodeValue(rule, kvm)
			if row.Action == canal.InsertAction || row.Action == canal.UpdateAction {
				docsMap[rule.ZincIndex] = append(docsMap[rule.ZincIndex], kvm)
			} else if row.Action == canal.DeleteAction {
				_, _, err := s.client.Document.Delete(context.Background(), rule.ZincIndex, strId).Execute()
				if err != nil {
					logs.Errorf("删除文档失败, index: %s, id: %s, err: %s", rule.ZincIndex, strId, errors.ErrorStack(err))
					alarm.EmergencyAlarm(err)
					return count
				}
				count++
			}
		}
	}

	if len(docsMap) > 0 {
		for index, docs := range docsMap {
			resp, _, err := bulk.Query(client.MetaJSONIngest{
				Index:   &index,
				Records: docs,
			}).Execute()
			if err != nil {
				ifZincsearchOpenapiErrLogBody(err)
				alarm.EmergencyAlarm(err)
				return count
			}
			logs.Infof("index: %s, resp message: %s, recourdCount: %d", index, *resp.Message, *resp.RecordCount)
			count += int64(*resp.RecordCount)
		}
	}

	return count
}

func (s *ZincSearchEndpoint) Close() {

}

func ifZincsearchOpenapiErrLogBody(err error) {
	var openapiErr *client.GenericOpenAPIError
	if errors.As(err, &openapiErr) {
		logs.Errorf("Zincsearch openapiErr body: %s", string(openapiErr.Body()))
	}
}
